package com.cy.improve.demos.web.mq;

import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.clients.admin.TopicListing;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;

import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ExecutionException;

@RestController
public class KafkaController {

    @Autowired
    private MessageProducer messageProducer;

    @PostMapping("/send")
    public String sendMessage(@RequestParam("message") String message) {
        messageProducer.sendMessage("my-topic", message);
        return "Message sent: " + message;
    }


    // 创建topic的接口，接收topic名称、分区数量和副本数量作为参数
    @PostMapping("/createTopic")
    public ResponseEntity<String> createTopic(@RequestParam("topicName") String topicName,
                                              @RequestParam("partitions") int partitions,
                                              @RequestParam("replicas") int replicas) {
        return messageProducer.createTopic(topicName, partitions, replicas);
    }

    @GetMapping("/topics")
    public ResponseEntity getTopicList() {
        try {
            List<Map<String, Object>> topicList = messageProducer.getTopicDetails();
            return new ResponseEntity<>(topicList, HttpStatus.OK);
        } catch (ExecutionException | InterruptedException e) {
            e.printStackTrace();
            return new ResponseEntity<>(HttpStatus.INTERNAL_SERVER_ERROR);
        }
    }

    @DeleteMapping("/deleteTopic")
    public ResponseEntity<String> deleteTopic(@RequestParam("topicName") String topicName) {
        try {
            messageProducer.deleteTopic(topicName);
            return new ResponseEntity<>("Topic " + topicName + " deleted successfully.", HttpStatus.OK);
        } catch (ExecutionException | InterruptedException e) {
            e.printStackTrace();
            return new ResponseEntity<>("Failed to delete topic: " + e.getMessage(), HttpStatus.INTERNAL_SERVER_ERROR);
        }
    }
}